﻿using System;
using System.Messaging;
using System.Runtime.InteropServices;
using System.Text;
using System.Threading;
using Interop.MSMQ;
using PSE.Framework.Threading;

namespace PSE.Framework.Messaging
{
    /// <summary>
    /// Creates and manages a connection with a message queue.
    /// </summary>
    public class MsmqReceiver : Msmq, IDisposable
    {
        #region Constants
        /// <summary>
        /// Identifies that the MessageQueue object cannot be read.
        /// </summary>
        public const int ERROR_CODE_MSMQ_CANT_READ = -800;
        /// <summary>
        /// 
        /// </summary>
        private const int ERROR_CODE_MSMQ_PAUSED = -799;
        /// <summary>
        /// Identifies a failure to read.
        /// </summary>
        public const int ERROR_CODE_MSMQ_FAILED_TO_READ = -798;
        /// <summary>
        /// Identifies an unexpected failure on read.
        /// </summary>
        public const int ERROR_CODE_UNEXPECTED_FAILURE_ON_READ = -797;
        /// <summary>
        /// Identifies an unexpected failure on connect.
        /// </summary>
        public const int ERROR_CODE_UNEXPECTED_FAILURE_ON_CONNECT = -796;
        /// <summary>
        /// Identifies a connection failure error.
        /// </summary>
        public const int ERROR_CODE_CONNECTION_FAILURE = -795;
        /// <summary>
        /// Identifies a queue doesn't exist error.
        /// </summary>
        public const int ERROR_CODE_QUEUE_DOESNT_EXIST = -794;
        /// <summary>
        /// Identifies an unexpected failure on thread running.
        /// </summary>
        public const int ERROR_CODE_UNEXPECTED_FAILURE_ON_THREAD_RUNNING = -793;
        /// <summary>
        /// Identifies an error when creating the msmq management object.
        /// </summary>
        public const int ERROR_CODE_CREATING_MSMQ_MANAGEMENT = -792;
        /// <summary>
        /// Identifies an error when reading the queue's size.
        /// </summary>
        public const int ERROR_CODE_READING_QUEUE_SIZE = -791;
        /// <summary>
        /// Identifies an error when reading the queue's message quantity.
        /// </summary>        
        public const int ERROR_CODE_READING_QUEUE_MESSAGE_QUANTITY = -790;
        /// <summary>
        /// Identifies an unexpected failure on check health status.
        /// </summary>
        public const int ERROR_CODE_UNEXPECTED_FAILURE_ON_CHECK_HEALTH_STATUS = -789;
        /// <summary>
        /// Identifies an unexpected failure on health status event.
        /// </summary>
        public const int ERROR_CODE_UNEXPECTED_FAILURE_ON_HEALTH_STATUS_EVENT = -788;

        // Tempo que espera uma mensagem chegar dentro do método Peek() ou Receive();
        private const int TIME_TO_WAIT_MESSAGES = 100000;

        // Tempo entre uma reconexão e outra em caso de erro
        private const int TIME_TO_CHECK_CONNECTION = 10000;

        // Tempo entre as tentativas de ler da fila, quando a fila está vazia
        private const int TIME_TO_RETRY_READ = 1000;

        // Tempo entre os updates de statística. 
        // OBS: Não aumentar muito pois nesse momento é verificado se a fila não foi apagada após se conectar
        private const int TTME_TO_UPDATE_STATISTICS = 2000;
        #endregion        

        #region Attributes
        private MSMQManagement _msmqManagement;
        private MessageQueue _messageQueue;
        private BinaryMessageFormatter _binaryMessageFormatter;        
        private TimeSpan _timeToWaitForMessage;
        private DateTime _lastHealthStatusChecked;
        private bool _keepRunning;        
        private bool _paused;
        private long _totalMsmqConnectionTries = 0;        
        private int _totalMessageSentPerInterval = 0;
        private int _messagesReceivedPerSecond;
        private bool _msmqManagementActivated;        
        #endregion        

        #region Properties

        /// <summary>
        /// Gets the number of messages received per second by the MsmqReceiver object.
        /// </summary>
        public int MessagesReceivePerSecond
        {
            get { return _messagesReceivedPerSecond; }
        }

        private bool IsRunning()
        {
            return _keepRunning;
        }        

        /// <summary>
        /// Gets the number of connection tries with the msmq object.
        /// </summary>
        public long TotalMsmqConnectionTries
        {
            get { return _totalMsmqConnectionTries; }
        }
                
        #endregion

        #region Delegates
        /// <summary>
        ///  Represents the method that will handle the MsmqReceiver.OnHealthStatusChanged event of a MsmqReceiver object.
        /// </summary>
        public delegate void HealthStatusChanged(MsmqReceiver msmqReceiver);
        /// <summary>
        /// Represents the method that will deliver the message.
        /// </summary>
        /// <param name="msg">The System.Messaging.Message object to be delivered.</param>        
        public delegate void MessageArrival(Message msg, MessageReceivedEventArgs e);
        #endregion        

        #region Events
        /// <summary>
        /// Occurs when the PSE.Framework.Messaging.MsmqReceiver object changed its state.
        /// </summary>
        public event HealthStatusChanged OnHealthStatusChanged;                        
        /// <summary>
        /// Occurs when the PSE.Framework.Messaging.MsmqReceiver object paused.
        /// </summary>
        public event EventHandler Paused;
        /// <summary>
        /// Occurs when the PSE.Framework.Messaging.MsmqReceiver object resumed.
        /// </summary>
        public event EventHandler Resumed;
        /// <summary>
        /// Occurs when the PSE.Framework.Messaging.MsmqReceiver received a message.
        /// </summary>
        public event MessageArrival MessageReceived;
        #endregion

        #region Constructor
        /// <summary>
        /// Initializes a new instance of the PSE.Framework.Messaging.MsmqReceiver class.
        /// </summary>
        /// <param name="path">The path of the message queue.</param>
        /// <param name="messageArrival">The method that will deliver the message.</param>
        public MsmqReceiver(string path)
        {
            if (string.IsNullOrEmpty(path))
                throw new ArgumentNullException("path");

            // Definir o path precisa ser a primeira linha do construtor, pois outras variáveis são definidas nesse momento
            this.Path = path;

            //Inicializações diversas
            _timeToWaitForMessage = new TimeSpan(TIME_TO_WAIT_MESSAGES);
            _lastHealthStatusChecked = DateTime.Now;
            _msmqManagementActivated = false;
            _paused = false;

            // Inicia a thread que vai se reconectar ao MSMQ em caso de queda e
            // ficará recebendo mensagens quando estas chegarem na fila.
            // esta thread fica viva até que _keepRunning seja false.
            Thread thr1 = new Thread(new ThreadStart(Run));
            thr1.Name = "MsmqReceiver - Receiver";
            thr1.Start();

            // Inicia a thread que vai monitorar a conexão com o MSMQ,
            // ficará fazendo update das estatísticas e disparando o evento HealthStatusChanged
            // esta thread fica viva até que _keepRunning seja false.
            Thread thr2 = new Thread(new ThreadStart(CheckMessageQueueHealthStatus));
            thr2.Name = "MsmqReceiver - Monitor";
            thr2.Start();            
        }
        #endregion        

        #region Methods
        private void Run()
        {
            // Declarações diversas
            Message msg;
            MessageReceivedEventArgs args = new MessageReceivedEventArgs();
            // Inicializações diversas
            _binaryMessageFormatter = new BinaryMessageFormatter();
            _keepRunning = true;            

            // Thread ficará rodando até _keepRunning = false
            // _keepRunning só será false na chamada do método Dispose. 
            while (_keepRunning)
            {
                try
                {
                    if (!_isConnected) //Na primeira vez não estará conectado, a partir daí, se estiver conectado, não faz nada
                    {
                        if (QueueExists(_queueName)) // Se a fila existir vai se conectar
                        {                            
                            try
                            {
                                // Se conecta ao MSMQ
                                _messageQueue = new MessageQueue(_path);
                                _isConnected = true;

                                HandleConnection();

                                while (_keepRunning && _isConnected) //Fica recebendo mensagens até o dispose ser chamado ou dar um problema de conexão
                                {
                                    try
                                    {
                                        if (!_paused) // Se alguém definiu que é para não ficar lendo do Msmq
                                        {                                           
                                            if (_messageQueue.CanRead) // Stream do MQ está liberado para leitura
                                            {
                                                if (MessageReceived != null) // Delegate que receberá a mensagem tem que ser diferente de null
                                                {
                                                    msg = _messageQueue.Peek(_timeToWaitForMessage); // Pega mensagem sem retirar da fila
                                                    msg.Formatter = _binaryMessageFormatter;
                                                    args.Result = false;
                                                    MessageReceived(msg, args);

                                                    if (args.Result) // Entrega mensagem e se o delegate retornar true, ou seja conseguiu processar, então retira da fila
                                                    {
                                                        _messageQueue.Receive(_timeToWaitForMessage);                                                        
                                                    }                                                    
                                                }

                                                _totalMessageSentPerInterval++; // Soma do contador que servirá para cálculo de msg/segundo
                                            }
                                            else // Se o stream do MSMQ não estiver disponível para leitura
                                            {
                                                // Provavelmente usuário não ter permissão na fila remota...

                                                _isConnected = false; //Tem que ser a primeira linha, para os que dependem dessa variável saber o quanto antes
                                                _totalMsmqConnectionTries++;

                                                _lastErrorMessage = "Msmq's CanRead Status did not allow the message to be sent.";
                                                _lastErrorCode = ERROR_CODE_MSMQ_CANT_READ;                                                

                                                Sleeper.Sleep(TIME_TO_CHECK_CONNECTION, IsRunning); // Dá um tempo para tentar se reconectar
                                            }
                                        }
                                        else // Se estiver pausado!
                                        {

                                            _lastErrorMessage = "Queue is paused.";
                                            _lastErrorCode = ERROR_CODE_MSMQ_PAUSED;                                            

                                            // Aguarda um tempo para ir na fila novamente
                                            Sleeper.Sleep(TIME_TO_RETRY_READ, IsRunning);
                                        }
                                    }
                                    catch (MessageQueueException messageQueueException)
                                    {
                                        // Se não for o erro de TimeOut do Peek ou Receive
                                        if (!(messageQueueException.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout))
                                        {
                                            _isConnected = false; //Tem que ser a primeira linha, para os que dependem dessa variável saber o quanto antes
                                            _totalMsmqConnectionTries++;

                                            _lastErrorMessage = "Error when trying to read queue..." + messageQueueException.Message;
                                            _lastErrorCode = ERROR_CODE_MSMQ_FAILED_TO_READ;

                                            HandleError(ERROR_CODE_MSMQ_FAILED_TO_READ, messageQueueException);
                                            HandleDisconnection();
                                        }
                                        else
                                        {
                                            // Se for erro de IOTimeOut, quer dizer que o Peek ou Receive 
                                            // ficaram sem conseguir ler a fila por um tempo, isso quer dizer
                                            // apenas que não chegou nada na fila o tempo configurado TIME_TO_WAIT_MESSAGES
                                            
                                            PSE.Framework.Threading.Sleeper.Sleep(TIME_TO_RETRY_READ, IsRunning); // Aguarda um tempo para ir na fila novamente
                                        }
                                    }
                                    catch (Exception ex)
                                    {
                                        _isConnected = false; //Tem que ser a primeira linha, para os que dependem dessa variável saber o quanto antes
                                        _totalMsmqConnectionTries++;

                                        _lastErrorMessage = "Unexpected error when trying to read MSMQ:" + ex.Message;
                                        _lastErrorCode = ERROR_CODE_UNEXPECTED_FAILURE_ON_READ;

                                        HandleError(ERROR_CODE_UNEXPECTED_FAILURE_ON_READ, ex);
                                        HandleDisconnection();
                                    }
                                }                                
                            }
                            catch (MessageQueueException messageQueueException)
                            {
                                // Provavelmente erro ao se conectar ao MSMQ
                                _isConnected = false; //Tem que ser a primeira linha, para os que dependem dessa variável saber o quanto antes
                                _totalMsmqConnectionTries++;

                                _lastErrorMessage = "Error when trying to connect to the MSMQ object:" + messageQueueException.Message;
                                _lastErrorCode = ERROR_CODE_CONNECTION_FAILURE;

                                HandleError(ERROR_CODE_CONNECTION_FAILURE, messageQueueException);
                                HandleDisconnection();
                            }
                            catch (Exception ex)
                            {
                                _isConnected = false; //Tem que ser a primeira linha, para os que dependem dessa variável saber o quanto antes
                                _totalMsmqConnectionTries++;

                                _lastErrorMessage = "Unexpected error when trying to connect to the MSMQ object:" + ex.Message;
                                _lastErrorCode = ERROR_CODE_UNEXPECTED_FAILURE_ON_CONNECT;

                                HandleError(ERROR_CODE_UNEXPECTED_FAILURE_ON_CONNECT, ex);
                                HandleDisconnection();
                            }
                        }
                        else //Se a fila não existir
                        {
                            _isConnected = false; //Tem que ser a primeira linha, para os que dependem dessa variável saber o quanto antes
                            _totalMsmqConnectionTries++;

                            _lastErrorMessage = "Queue doesn't exist or cannot be accessed.";
                            _lastErrorCode = ERROR_CODE_QUEUE_DOESNT_EXIST;

                            HandleError(ERROR_CODE_QUEUE_DOESNT_EXIST, new Exception(_lastErrorMessage));
                            HandleDisconnection();
                        }
                    }
                }
                catch (Exception ex)
                {
                    _isConnected = false; //Tem que ser a primeira linha, para os que dependem dessa variável saber o quanto antes
                    _totalMsmqConnectionTries++;

                    _lastErrorMessage = "It was impossible to detect the error...";
                    _lastErrorCode = ERROR_CODE_UNEXPECTED_FAILURE_ON_THREAD_RUNNING;

                    HandleError(ERROR_CODE_UNEXPECTED_FAILURE_ON_THREAD_RUNNING, ex);
                    HandleDisconnection();
                }                

                // Dá um tempo para checar a conexão e se for o caso tentar se conectar novamente
                Sleeper.Sleep(TIME_TO_CHECK_CONNECTION, IsRunning);
            }            
        }

        /// <summary>
        /// Pauses the MsmqReceiver object.
        /// </summary>
        public void PauseMsmqReceiver()
        {
            _paused = true;

            if (Paused != null)
                Paused(this, EventArgs.Empty);
        }

        /// <summary>
        /// Resumes the MsmqReceiver object.
        /// </summary>
        public void ResumeMsmqReceiver()
        {
            _paused = false;

            if (Resumed != null)
                Resumed(this, EventArgs.Empty);
        }

        private bool StartMsmqManagement()
        {
            bool ret;
            try
            {
                _msmqManagement = null;
                _msmqManagement = new MSMQManagement();
                _msmqManagement.Init(ref _machineNameObject, ref _queueNameObject, ref _formatNameObject);
                ret = true;                
            }
            catch (Exception ex)
            {
                _lastErrorMessage = "Error when trying to create Interop.MSMQManagement object.";
                _lastErrorCode = ERROR_CODE_CREATING_MSMQ_MANAGEMENT;
                ret = false;

                HandleError(ERROR_CODE_CREATING_MSMQ_MANAGEMENT, ex);
            }

            return ret;
        }

        private void CheckMessageQueueHealthStatus()
        {
            _keepRunning = true;

            // Thread ficará rodando até _keepRunning = false
            // _keepRunning só será false na chamada do método Dispose.
            
            while (_keepRunning)
            {
                try
                {
                    // Monitora caso a fila fique indisponível após conectado
                    if (!QueueExists(_queueName))
                    {
                        _isConnected = false;
                        _totalMsmqConnectionTries++;

                        _lastErrorMessage = "Queue doesn't exist or cannot be accessed.";
                        _lastErrorCode = ERROR_CODE_QUEUE_DOESNT_EXIST;

                        HandleError(ERROR_CODE_QUEUE_DOESNT_EXIST, new Exception(_lastErrorMessage));
                        HandleDisconnection();
                    }                    

                    // Reativa o objeto de gerenciamento do MSMQ caso esse seja desativado, 
                    if (!_msmqManagementActivated)
                    {
                        _msmqManagementActivated = StartMsmqManagement();
                    }                    

                    // Lê tamanho da fila em Bytes
                    try
                    {
                        if (_msmqManagementActivated)
                        {
                            _queueSizeBytes = long.Parse(_msmqManagement.BytesInQueue.ToString());                            
                        }                        
                    }
                    catch (Exception ex)
                    {
                        // Se der erro, é porque o objeto de gerenciamento do MSMQ deu problema
                        _queueSizeBytes = -1;
                        _msmqManagementActivated = false; // Define o objeto de gerenciamento do MSMQ  como desativado

                        _lastErrorMessage = "It was not possible to read the queue's size.";
                        _lastErrorCode = ERROR_CODE_READING_QUEUE_SIZE;

                        HandleError(ERROR_CODE_READING_QUEUE_SIZE, ex);                        
                    }

                    // Lê tamanho número de mensagens da fila
                    try
                    {
                        if (_msmqManagementActivated)
                        {
                            _queuedMessages = _msmqManagement.MessageCount;                            
                        }                        
                    }
                    catch (Exception ex)
                    {
                        // Se der erro, é porque o objeto de gerenciamento do MSMQ deu problema
                        _queuedMessages = -1;
                        _msmqManagementActivated = false; // Define o objeto de gerenciamento do MSMQ  como desativado

                        _lastErrorMessage = "It was not possible to read the queue's message number.";
                        _lastErrorCode = ERROR_CODE_READING_QUEUE_MESSAGE_QUANTITY;

                        HandleError(ERROR_CODE_READING_QUEUE_MESSAGE_QUANTITY, ex);                        
                    }

                    // Calcula o número de mensagens recebidas por segundo
                    _messagesReceivedPerSecond = CalculateMessageNumberPerSecond();

                    // Dispara o evento com os dados de estatística do componente
                    ThrowEventOnHealthStatusChanged();
                    
                    // Aguarda um tempo até checar a saúde da fila e coletar as estatísticas novamente
                    Sleeper.Sleep(TTME_TO_UPDATE_STATISTICS, IsRunning);
                }
                catch (Exception ex)
                {
                    _lastErrorMessage = "Unidentified error on CheckMessageQueueHealthStatus() sub.";
                    _lastErrorCode = ERROR_CODE_UNEXPECTED_FAILURE_ON_CHECK_HEALTH_STATUS;

                    HandleError(ERROR_CODE_UNEXPECTED_FAILURE_ON_CHECK_HEALTH_STATUS, ex);                    
                }
            }
        }

        private void ThrowEventOnHealthStatusChanged()
        {
            try
            {
                if (OnHealthStatusChanged != null) // Caso alguém tenha assinado o evento
                    OnHealthStatusChanged(this);
            }
            catch (Exception ex)
            {
                // Caso dentro do componente que assinou o evento dê um erro
                _lastErrorMessage = "Error when throwing OnHealthStatusChanged() event.";
                _lastErrorCode = ERROR_CODE_UNEXPECTED_FAILURE_ON_HEALTH_STATUS_EVENT;

                HandleError(ERROR_CODE_UNEXPECTED_FAILURE_ON_HEALTH_STATUS_EVENT, ex);                
            }
        }

        private int CalculateMessageNumberPerSecond()
        {
            int seconds;

            // Calcula o número de segundos entre a última coleta e a coleta atual
            seconds = DateTime.Now.Subtract(_lastHealthStatusChecked).Seconds;

            if (seconds == 0)
                seconds = TIME_TO_CHECK_CONNECTION / 1000;

            if (_totalMessageSentPerInterval >= seconds)
                _messagesReceivedPerSecond = _totalMessageSentPerInterval / seconds;
            else
                _messagesReceivedPerSecond = _totalMessageSentPerInterval;

            // Zera o contador que será usado na próxima coleta
            // Esse contador é incrementado dentro do Run()
            // a cada mensagem recebida
            _totalMessageSentPerInterval = 0;

            // Define o horário dessa verificação, para ser usada na próxima chamada da função e calcular os segundos entre uma chamada e outra
            _lastHealthStatusChecked = DateTime.Now;
            
            return _messagesReceivedPerSecond;
        }

        /// <summary>
        /// Releases all resources.
        /// </summary>
        public void Dispose()
        {
            // Tem que ser as duas primeiras linhas, para os que dependem dessa variável saber o quanto antes
            // _keepRunning = false tem que vir antes de _connected = false
            // caso contrario corre-se o risco do status mudar para conectado por a thread ainda estar aberta.
            _keepRunning = false; // Determina que as Thread abertas podem ser fechadas...
            _isConnected = false; // Determina que não estamos mais conectados...
            
            try
            {                
                if (_messageQueue != null)
                    _messageQueue.Close();

                if (_msmqManagement != null)
                    Marshal.ReleaseComObject(_msmqManagement);

                _messageQueue = null;
                _msmqManagement = null;
            }
            catch (Exception ex)
            {
                StringBuilder error = new StringBuilder();
                error.Append("Error when disposing the object... \\n");
                error.Append(string.Concat(ex.Message, " \\n"));
                error.Append(string.Concat(ex.StackTrace, " \\n"));

                HandleError(0, new Exception(error.ToString()));
            }                        
        }       
     
        #endregion                        
    }
}
